package com.sdg.streaming

import breeze.numerics.log
import com.mongodb.BasicDBObject
import com.mongodb.casbah.commons.MongoDBObject
import com.mongodb.casbah.{MongoClient, MongoClientURI, MongoCollection}
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.SparkSession
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import redis.clients.jedis.Jedis

import scala.collection.mutable

/**
  * 实时推荐模块
  */
object StreamingRecomender {
  val minSimilarity = 0.7
  val MAX_REC_COUNT = 50
  //获取当前用户m次评分
  val USER_RECENTLY_RATING_COUNT = 10
  val SIMILAR_MOVIE_COUNT = 20

  //实时推荐电影表
  val MONGODB_STREAM_RECS_COLLECTION: String = "StreamRecs"

  val MONGO_URI: String = "mongodb://vm2:27017/recom3";

  val MONGODB_DATABASE: String = "recom3"

  val KAFKA_BOOTSTRAP_SERVERS: String = "vm0:9092,vm1:9092,vm2:9092"
  // val KAFKA_BOOTSTRAP_SERVERS: String = "vm2:9092"
  //电影相似度矩阵
  val MONGODB_MOVIE_RECS_COLLECTION = "MoviesRecs"

  val MONGO_DRVIVE_CLASS = "com.mongodb.spark.sql"

  val KAFKA_OUTPUT_TOPIC = "source"

  //程序主入口
  def main(args: Array[String]): Unit = {
    //封装配置信息
    val conf = Map(
      "spark.cores" -> "local[2]",
      "mongo.uri" -> MONGO_URI,
      "mongo.db" -> MONGODB_DATABASE
    )

    //创建一个Sparkconf
    val sparkConf: SparkConf = new SparkConf().setAppName("StreamingRecomender").setMaster("local[2]")
      .set("spark.executor.memory", "6G").set("spark.driver.memory", "3G")

    //创建一个sparkSession
    val spark: SparkSession = SparkSession.builder().config(sparkConf).getOrCreate()
    //创建一个kafka的链接linux
    //kafka消费者配置
    val kafkaParam = Map(
      "bootstrap.servers" -> KAFKA_BOOTSTRAP_SERVERS, //用于初始化链接到集群的地址
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      //消费者组(随便起名)
      "group.id" -> "test",
      //如果没有初始化偏移量或者当前的偏移量不存在任何服务器上，可以使用这个配置属性
      //可以使用这个配置，latest自动重置偏移量为最新的偏移量
      "auto.offset.reset" -> "latest",
      //如果是true，则这个消费者的偏移量会在后台自动提交
      "enable.auto.commit" -> (false: java.lang.Boolean)
    )
    //创建DStream，返回接收到的输入数据
    val sc = spark.sparkContext
    //5秒计算一次
    val ssc = new StreamingContext(sc, Seconds(2))

    import spark.implicits._
    //获取电影相似矩阵
    val movieRecs = spark.read
      .option("uri", MONGO_URI)
      .option("collection", MONGODB_MOVIE_RECS_COLLECTION) //电影相似度矩阵  数据量太大出现卡死现象
      .format(MONGO_DRVIVE_CLASS)
      .load()
      //列和行的选择
      .select($"mid", $"recommendations")
      .as[MovieRecommendation]
      .map { rec => (rec.mid, rec.recommendations.map(x => (x.rid, x.r)).toMap) }
      .rdd.cache()
    println(movieRecs.collect().toBuffer)
    //每个电影与其他电影的相似度，HASH[电影Id1, HASH[电影Id2, Id1与Id2相似度]]
    //把datase转换成map  并且key进行了一次去重[mid:[mid:score,mid:score]]
    val movie2movieSimilarity = movieRecs.collectAsMap()
    //每个电影的最相似的K个电影，HASH[电影Id, 相似的K个电影Ids]
    //[mid,[mid1,mid2,mid3]]
    val topKMostSimilarMovies = movieRecs.map(rec => (rec._1, rec._2.map(_._1).toArray)).collectAsMap()

    //电影间相似度HASH的广播
    val bMovie2movieSimilarity = ssc.sparkContext.broadcast(movie2movieSimilarity)
    //最相似电影HASH的广播
    val bTopKMostSimilarMovies = ssc.sparkContext.broadcast(topKMostSimilarMovies)


    //创建DStream，返回接收到的输入数据
    //UID|MID|SCORCE|TIMESTAMP
    val stream = KafkaUtils.createDirectStream[String, String](ssc, LocationStrategies.PreferConsistent, ConsumerStrategies.Subscribe[String, String](Array(KAFKA_OUTPUT_TOPIC), kafkaParam))

    val ratingStream = stream.map {
      case msg =>
        println(msg.value() + "-------------")
        val arr: Array[String] = msg.value().split("\\|")
        (arr(0).toInt, arr(1).toInt, arr(2).toDouble, arr(3).toInt)
    }


    ratingStream.foreachRDD(rdd => {
      rdd.map {
        case (userId, movieId, score, timestamp) => {
          println("输出提示:<<>>>>>>>>>>>>>>>>>>>>>>>>>")
          //逻辑运算
          //从Redis中获取当前用户近期M次评分记录(这里redis必须启动)
          val recentRatings: Array[(Int, Double)] = getUserRecentRatings(SingleConnectionClient.getRedisClient(), USER_RECENTLY_RATING_COUNT, userId, movieId, score)
          //val recentRatings=Array((1,1.2d))
          val mongoCollection: MongoCollection = SingleConnectionClient.getMongoDBCollection("Rating")
          //获取当前评分电影相似的K个备选电影
          val candidateMovies = getSimilarMovies(bTopKMostSimilarMovies.value, mongoCollection, movieId, userId, SIMILAR_MOVIE_COUNT)
          //为备选电影根据计算公式计算推荐的优先级
          val updatedRecommends = createUpdatedRatings(bMovie2movieSimilarity.value, recentRatings, candidateMovies)
          //当前推荐与往期推荐进行Merge结果回写到MONGODB
          updateRecommends2MongoDB(SingleConnectionClient.getMongoDBCollection(MONGODB_STREAM_RECS_COLLECTION), updatedRecommends, userId, timestamp)
        }
      }.count() //触发操作
    })
    //启动sreaming的程序
    ssc.start()
    ssc.awaitTermination()

  }

  /**
    * 将备选电影的预期评分合并后回写到MONGODB中
    *
    * @param collection
    * @param newRecommends
    * @param userId
    * @param startTimeMillis
    * @return
    */
  def updateRecommends2MongoDB(collection: MongoCollection, newRecommends: Array[(Int, Double)], userId: Int, startTimeMillis: Long): Unit = {

    val query = MongoDBObject("uid" -> userId)
    //根虎uid 查出来
    collection.findOne(query) match {
      //如果mongo中没有的话我们就直接写入
      case None => {
        insertStreamRecs(collection, newRecommends, userId, startTimeMillis)
      }
      //如果有的话
      case oldRecs => {
        val oldMap = scala.collection.mutable.Map[Int, Double]()
        //请说明 oldRecs 是上一次计算出来还未推荐的结果
        oldRecs.toArray.apply(0).asInstanceOf[BasicDBObject]
          .getString("recs")
          .split("\\|")
          .map { item =>
            val attr = item.split(",");
            if (attr.length >= 2) {
              oldMap += (attr(0).toInt -> attr(1).toDouble)
            }
            oldMap
          }

        for (item <- newRecommends) {
          if (oldMap.contains(item._1.toInt)) {
            if (oldMap.get(item._1.toInt).get < item._2.toDouble) {
              oldMap.put(item._1.toInt, item._2.toDouble)
            }
          } else {
            oldMap.put(item._1.toInt, item._2.toDouble)
          }
        }

        collection.remove(query)

        //整体的话做一个排序
        val newArray = oldMap.toList.sortWith(_._2 < _._2).slice(0, MAX_REC_COUNT).toArray
        //再次写入到我们的数据库中
        insertStreamRecs(collection, newArray, userId, startTimeMillis)
      }
    }
  }

  /**
    *
    * @param collection
    * @param newRecommends
    * @param userId
    * @param startTimeMillis
    */
  def insertStreamRecs(collection: MongoCollection, newRecommends: Array[(Int, Double)], userId: Int, startTimeMillis: Long): Unit = {
    val toInsert = MongoDBObject("uid" -> userId,
      "recs" -> newRecommends.map(item => Recommendation(item._1, item._2)),
      "timedelay" -> (System.currentTimeMillis() - startTimeMillis).toDouble / 1000)
    //数据写入到mongo中
    collection.insert(toInsert)
  }

  import scala.collection.JavaConversions._

  /**
    * 从Redis中获取当前用户最近K次评分，返回 Buffer[(Int,Double)]
    *
    * @param jedis   Redis客户端
    * @param K       获取当前用户K次评分
    * @param userId  用户ID
    * @param movieId 新评分的电影ID
    * @param score   新评分的电影评分
    * @return 最近K次评分的数组，每一项是<movieId, score>
    */
  def getUserRecentRatings(jedis: Jedis, K: Int, userId: Int, movieId: Int, score: Double): Array[(Int, Double)] = {
    //取出指定键上的从0开始的第个k元素 k+1个元素
    jedis.lrange(userId.toString, 0, K).map {
      line =>
        val attr = line.split(":")
        (attr(0).toInt, attr(1).toDouble)
    }.toArray
  }

  /**
    * 从广播变量中获取movieId最相似的K个电影，并通过MONGODB来过滤掉已被评分的电影
    *
    * @param mostSimilarMovies          电影相对应的相似矩阵
    * @param collectionForRatingRecords MongDB数据库连接
    * @param movieId                    当前评分的MovieID
    * @param userId                     当前评分的UserID
    * @param K
    * @return 返回相似的电影ID数组
    */
  def getSimilarMovies(mostSimilarMovies: scala.collection.Map[Int, Array[Int]], collectionForRatingRecords: MongoCollection, movieId: Int, userId: Int, K: Int): Array[Int] = {
    //获取之前的电影相似度矩阵
    val similarMoviesBeforeFilter = mostSimilarMovies.getOrElse(movieId, Array[Int]())
    //数据
    val query = MongoDBObject("uid" -> userId)
    //collectionForRatingRecords : MongDB数据库连接
    //从mongo 中查出已经有偏好的电影
    val hasRated: Set[Int] = collectionForRatingRecords.find(query)
      .toArray
      .map(_.get("mid").toString.toInt).toSet
    //把已经看过的电影去掉
    similarMoviesBeforeFilter.filter(hasRated.contains(_) == false)

  }

  /**
    * 核心算法，计算每个备选电影的预期评分
    *
    * @param simiHash        电影之间的相似度矩阵
    * @param recentRatings   用户最近评分的K个电影集合
    * @param candidateMovies 当前评分的电影的备选电影集合
    * @return 备选电影预计评分的数组，每一项是<movieId, maybe_rate>
    */
  def createUpdatedRatings(simiHash: scala.collection.Map[Int, scala.collection.Map[Int, Double]], recentRatings: Array[(Int, Double)], candidateMovies: Array[Int]): Array[(Int, Double)] = {
    val allSimilars = mutable.ArrayBuffer[(Int, Double)]()
    val increaseCounter = mutable.Map[Int, Int]()
    val reduceCounter = mutable.Map[Int, Int]()
    //遍历备选电影集合和最近评分的电影集合
    for (cmovieId <- candidateMovies; (rmovieId, rate) <- recentRatings) {
      //(电影之间的相似度矩阵,最近的电影id,候选的电影id) 返回电影相似度
      val sim = getSimilarityBetween2Movies(simiHash, rmovieId, cmovieId)
      //如果电影的相似度大于我们的阈值
      if (sim > minSimilarity) {
        allSimilars += ((cmovieId, sim * rate))
        //如果电影的评分是大于3的
        if (rate >= 6.0) {
          increaseCounter(cmovieId) = increaseCounter.getOrElse(cmovieId, 0) + 1
        } else {
          reduceCounter(cmovieId) = reduceCounter.getOrElse(cmovieId, 0) + 1
        }
      }
    }
    allSimilars.toArray.groupBy {
      case (movieId, value) => movieId
    }
      .map { case (movieId, simArray) =>
        (movieId, simArray.map(_._2).sum / simArray.length + log(increaseCounter.getOrElse[Int](movieId, 1)) - log(reduceCounter.getOrElse[Int](movieId, 1)))
      }.toArray
  }

  /**
    * 从广播变量中获取movieId1与movieId2的相似度，不存在、或movieId1=movieId2视为毫无相似，相似度为0
    *
    * @param simHash  电影相似度矩阵
    * @param movieId1 最近电影id
    * @param movieId2 候选的电影id
    * @return movieId1与movieId2的相似度
    */
  def getSimilarityBetween2Movies(simHash: scala.collection.Map[Int, scala.collection.Map[Int, Double]], movieId1: Int, movieId2: Int): Double = {

    //对两个电影id进行一个排序,小的放在第一个,大的放在第二个
    val (smallerId, biggerId) = if (movieId1 < movieId2) (movieId1, movieId2) else (movieId2, movieId1)
    //如果两个电影是同一个电影的话我们,就设为二者的相似度为0
    if (smallerId == biggerId) {
      return 0.0
    }

    //从电影矩阵中去获取相似度,如果没有的话,就设为0
    simHash.get(smallerId) match {
      case Some(subSimHash) =>
        subSimHash.get(biggerId) match {
          case Some(sim) => sim
          case None => 0.0
        }
      case None => 0.0
    }
  }

}

//单例类获取连接，逃避序列化问题，且一个JVM只有一个连接者，提高性能
object SingleConnectionClient extends Serializable {
  lazy val mongoClient = MongoClient(MongoClientURI("mongodb://vm2:27017/recom3"))

  lazy val jedis = new Jedis("vm2")

  //获取mongodb的链接
  def getMongoDBCollection(collectionName: String): MongoCollection = {
    val collection: MongoCollection = mongoClient("recom3")(collectionName)
    collection
  }

  //获取redis的客户端
  def getRedisClient(): Jedis = {
    jedis
  }
}

/**
  * 推荐项目
  *
  * @param rid 项目ID
  * @param r   推荐分数
  */
case class Recommendation(rid: Int, r: Double)

/**
  * 电影相似推荐
  *
  * @param mid             电影ID
  * @param recommendations 相似的电影集合
  */
case class MovieRecommendation(mid: Int, recommendations: Seq[Recommendation])
